/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package mapred.wordcount;

import mapred.cuspartitioner.CusPartitioner;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

import java.io.IOException;
import java.util.StringTokenizer;

/**
 * 导出Jar包后的执行方式：
 * hadoop jar ~/Hadoop-CH6.jar  mapred.wordcount.WordCount ./input ./output/wordcount1
 * hadoop jar  jar包所在的路径    要执行的类的全名称           输入文件夹  输出文件夹
 */
public class WordCount {

    public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> {

        // 做为全局的临时变量使用，并且固定值是1
        private final static IntWritable one = new IntWritable(1);
        private Text word = new Text();

        public void map(Object key, Text value, Context context) throws IOException, InterruptedException {

//            Deer Bear River
//            Car Car River
//            Deer Car Bear
            // 假定：上述3行在一个input.txt文件中
            // 上述3行，每一行的行号是key的值
            // 每一行的内容是value的值
            // StringTokenizer类的说明：自动拆解一个字符串，按照\t以及空格等实现分词
            // 例如：Deer Bear River， 会被分为3个，也就是下面的循环会执行3次
            // Text类型是一个封装的对象，要获得它的值，需要使用toString方法
            // WordCount的算法核心之一：分词
            StringTokenizer itr = new StringTokenizer(value.toString());

            // 循环遍历被分词后的每个单词
            // hasMoreTokens()，如果遍历时下一个值还存在，返回true
            while (itr.hasMoreTokens()) {

                // 将每个单词的值放入Text类型的临时变量中，Text也是实现了Writable接口的
                // nextToken()，返回当前迭代位置的单词
                // WordCount的算法核心之二：将单词作为Key放入Mapper的输出
                word.set(itr.nextToken());
                // 必须使用context对象，将我们处理完的所有单词放入Mapper的输出，也就是给了Reducer
                //WordCount的算法核心之三：将每个分出的单词，设置它的value=1，方便Reducr中对每一个合并后的代词进行计数
                // Mapper的输出将是：
                // Deer 1
                // Bear 1
                // River 1
                context.write(word, one);
            }
        }
    }

    public static class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
        private IntWritable result = new IntWritable();

        // 从Mapper中输出的数据，已经经过Combiner进行了合并，所以，对于Key=Bear的Key来说，它的值的形式是：
        // Key=Bear Value=[1,1](可迭代类型，Iterable)
        // 对于Mapper输出的每一个Key，这个方法都会执行一次
        // 按照上面的文件输入，这里会执行4次：Bear、Car、Dear、River（按照这个顺序来执行，按照Key的字典序）
        public void reduce(Text key, Iterable<IntWritable> values,
                           Context context
        ) throws IOException, InterruptedException {

            // 合计每个Key的Value中的值的和
            int sum = 0;

            // WordCount的算法核心之四：遍历同一个Key的Value的数量
            for (IntWritable val : values) {
                sum += val.get();
            }
            // 将计算的结果放入临时变量：result
            result.set(sum);
            // WordCount的算法核心之五：将每个单词和计算之后的结果，利用context输出到HDFS文件系统中
            context.write(key, result);
        }
    }

    public static void main(String[] args) throws Exception {

        // 创建当前任务的配置信息，Configuration对象会从多个Hadoop的配置文件中获取配置信息
        Configuration conf = new Configuration();

        // L71-L76: 将我们关心的最后2个以上的参数提取出来。
        // hadoop jar XXXXXX.jar WordCount input1 input2 input3 output
        // input1 input2 input3 output == otherArgs
        // 提取之后的参数：第一个以后的是输入文件夹；最后个是输出文件夹
        String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
        if (otherArgs.length < 2) {
            System.err.println("Usage: mapred.wordcount.WordCount <in> [<in>...] <out>");
            System.exit(2);
        }

        // 通过单例模式获取Job的实例对象，并且，给Job起名为：【word count】
        Job job = Job.getInstance(conf, "word count");
        // 设置Jar包的信息，即设置Jar包中的启动类
        job.setJarByClass(WordCount.class);
        // 设置Mapper类，这个类是我们继承Mapper父类后自己实现的代码
        // 也是MapReduce模型中的Mapper代码，需要程序员自己实现
        job.setMapperClass(TokenizerMapper.class);
        // 设置Combiner类，此类可以跟Reducer类相同
        job.setCombinerClass(IntSumReducer.class);

        // 设置Reducer类，这个类是我们继承Reducer父类后自己实现的代码
        // 也是MapReduce模型中的Reducer代码，需要程序员自己实现
        job.setReducerClass(IntSumReducer.class);

        // 设置MapReduce的输出Key的类型，注意：这是最终结果的Key，即：Part-R-0000文件中的key
        job.setOutputKeyClass(Text.class);
        // 设置MapReduce的输出Value的类型，注意：这是最终结果的Value，即：Part-R-0000文件中的Value
        job.setOutputValueClass(IntWritable.class);

        // TODO: 设置了，但是目前不起作用
        job.setPartitionerClass(CusPartitioner.class);

        // 设置一次启动多少Reducer实例
        job.setNumReduceTasks(2);
        // Line99-101: 给MapReduce添加输入路径
        // 将所有输入参数的前N-1个路径当做输入路径
        for (int i = 0; i < otherArgs.length - 1; ++i) {
            FileInputFormat.addInputPath(job, new Path(otherArgs[i]));
        }

        // 通过conf对象获取文件系统对象，如果输出路径已经存在，先将其删除
        // 注意：练习过程中可以这样做，实际开发中，尽量避免删除操作
        FileSystem fs = FileSystem.get(conf);
        if (fs.exists(new Path(otherArgs[otherArgs.length - 1]))) {
            fs.delete(new Path(otherArgs[otherArgs.length - 1]), true);
        }
        // 设置MapReduce的输出路径，MapReduce框架要求，此路径的文件夹应该是不存在的
        FileOutputFormat.setOutputPath(job, new Path(otherArgs[otherArgs.length - 1]));

        // 通过Job的waitForCompletion启动我们的MapReduce任务
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}
